package lab2;

import java.io.*;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

public class SR {
    private static InetAddress fromHost; //获取发送给本机的ip地址
    private static int fromPort;//获取发送给本机的端口号

    /**
     * 从文件中读取数据
     * @param file 文件
     * @return 读取出的字节数组
     * @throws IOException IO异常
     */
    public static byte[] getByteArray(File file) throws IOException {
        FileInputStream fileInputStream = new FileInputStream(file);
        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
        byte[] buffer = new byte[Configuration.maxDataLength]; //分配块大小
        int length;//读出的实际块大小
        while ((length = fileInputStream.read(buffer)) != -1) {
            byteArrayOutputStream.write(buffer, 0, length);
        }
        return byteArrayOutputStream.toByteArray();
    }

    /**
     * 发送数据给目的主机中的目的端口号
     * @param socket 发送方套接字
     * @param sendBytes 需要发送的字节数组
     * @param targetHost 目的ip
     * @param targetPort 目的端口号
     * @throws IOException IO异常
     */
    public static void send(DatagramSocket socket, byte[] sendBytes, InetAddress targetHost, int targetPort) throws IOException {
        int dataGramNumber = sendBytes.length / Configuration.maxDataLength; //需要发送的数据报的个数
        int lastDataGramLength = sendBytes.length % Configuration.maxDataLength; //最后一个数据报的大小
        Map<Integer, Integer> ackMap = new ConcurrentHashMap<>(); //线程安全的Hashmap
        int base = 0; //发送窗口的基数
        int next = 0; //下一个需要发送的数据报的序号
        byte[] toSendBytes; //用于构造每个需要发送的数据报
        ReceiveACK receiveACK = new ReceiveACK(socket, ackMap);//构造用于接收ACK的线程
        receiveACK.start(); //开始接收ack
        while (next <= dataGramNumber || ackMap.size() != 0) {
            int oldBase = base;
            int oldSize = ackMap.size();
            for (int i = 0; i < oldSize; i++) { //滑动窗口
                Integer now = (oldBase + i) % Configuration.maxACKNumber;
                if (ackMap.get(now) == -1) //如果没收到该ack，则不能再继续滑动窗口
                    break;
                base = (base + 1) % Configuration.maxACKNumber;//窗口滑动一格
                ackMap.remove(now);//删除已经在base左侧的ack
            }
            int canSend = Configuration.sendWindowSize - ackMap.size();//计算还可以发送多少数据报
            int ackNumber = (base + ackMap.size()) % Configuration.maxACKNumber; //最大长度
            for (int i = 0; i < canSend; i++) {
                if (next > dataGramNumber)
                    break;
                int sequence = (ackNumber + i) % Configuration.maxACKNumber; //数据报的序列号
                int sendLength = next < dataGramNumber ? Configuration.maxDataLength : lastDataGramLength;
                int end = next < dataGramNumber ? 1 : 0; //标记所有报文是否全部到达
                //构造报文；报文格式：0-2字节：报文中数据的长度；3字节：报文序号；4-1028字节：报文携带的数据
                toSendBytes = new byte[sendLength + 4]; //预分配空间
                toSendBytes[0] = (byte) ((sendLength >> 8) & 0xff); //数据总长最多1024，即至少需要10位来进行编码，构造报文时按16位（2Byte）进行构造
                toSendBytes[1] = (byte) (sendLength & 0xff); //报文第一二字节总和表示数据长度
                toSendBytes[2] = (byte) sequence; //报文序号
                System.arraycopy(sendBytes, next * Configuration.maxDataLength, toSendBytes, 3, sendLength);///拷贝需要发送的数据到数据报
                toSendBytes[sendLength + 3] = (byte) end; //最后一个字节标记所有报文是否全部到达
                DatagramPacket datagramPacket = new DatagramPacket(toSendBytes, sendLength + 4, targetHost, targetPort);

                if (Math.random() > Configuration.sendLoss) { //成功发送，否则不发送该数据报（发送端丢包）
                    System.out.println("发送了： " + sequence);
                    socket.send(datagramPacket);
                }else {
                    System.out.println("丢包： " + sequence);
                }
                next++;
                ackMap.put(sequence, -1);//标记该序号未收到ack
                new SRTimeOut(ackMap, socket, datagramPacket, sequence).start();//开始计时
            }
        }
    }

    /**
     * 接收方接收数据
     * @param socket 接收方的套接字
     * @return 接收到的数据
     * @throws IOException IO异常
     */
    public static byte[] receive(DatagramSocket socket) throws IOException {
        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
        byte[] receiveBytes = new byte[Configuration.maxDataLength+4];//报文大小
        int length;
        int base = 0;
        Map<Integer, byte[]> cache = new HashMap<>();
        while (true) {
            DatagramPacket packet = new DatagramPacket(receiveBytes, receiveBytes.length);
            try {
                socket.receive(packet); //接收数据报
            } catch (IOException e) {
                break;
            }
            //解析报文；报文格式：0-2字节：报文中数据的长度；3字节：报文序号；4-1028字节：报文携带的数据
            length = receiveBytes[0] & 0xff;
            length = length << 8 | receiveBytes[1] & 0xff; //还原报文中数据的长度
            int sequence = receiveBytes[2] & 0xff; //接收到的序号
            fromHost = packet.getAddress(); //获取主机号
            fromPort = packet.getPort();//获取端口号
            byte[] receiveAck = new byte[2];//构造ack报文
            receiveAck[0] = receiveBytes[2]; //接收到的序号
            receiveAck[1] = receiveBytes[length + 3];//标记整个数据传输是否结束
            byte[] arrivalData = Arrays.copyOfRange(receiveBytes, 3, 3 + length);//拷贝到达的数据报
            int oldWindowLow = (base - Configuration.receiveWindowSize + Configuration.maxACKNumber) % Configuration.maxACKNumber;//上一个窗口的最小序号
            int oldWindowHigh = (base + Configuration.maxACKNumber - 1 ) % Configuration.maxACKNumber;//上一个窗口的最大序号
            int newWindowLow = base;//本窗口的最小序号
            int newWindowHigh = (base + Configuration.receiveWindowSize - 1) % Configuration.maxACKNumber; //本窗口的最大序号
            if (((oldWindowLow > oldWindowHigh) && (sequence >= oldWindowLow || sequence <= oldWindowHigh)) || (sequence >= oldWindowLow && sequence <= oldWindowHigh)) {
                DatagramPacket ackPocket = new DatagramPacket(receiveAck, receiveAck.length, fromHost, fromPort);  //在老的发送窗口范围内，不接收数据报，但发送ack
                socket.send(ackPocket);
                System.out.println("重复收到了：" + sequence);
                continue;
            }
            if (((newWindowLow > newWindowHigh) && (sequence >= newWindowLow || sequence <= newWindowHigh)) || (sequence >= newWindowLow && sequence <= newWindowHigh)) {
                //在现在的发送窗口范围内，正确接收数据报，发送ack
                if (sequence == base) { //连续的滑动接收窗口
                    byteArrayOutputStream.write(arrivalData, 0, arrivalData.length);
                    base = (base + 1) % Configuration.maxACKNumber;
                    while (cache.get(base) != null) {
                        byteArrayOutputStream.write(cache.get(base), 0, cache.get(base).length);
                        cache.remove(base); //不remove也行，下次到这个序列号就更新了
                        base = (base + 1) % Configuration.maxACKNumber;
                    }
                } else {
                    cache.put(sequence, arrivalData); //不连续的加入缓存
                }
                DatagramPacket ackPocket = new DatagramPacket(receiveAck, receiveAck.length, fromHost, fromPort);  //发送ack
                System.out.println("收到了：" + sequence);
                if (Math.random() > Configuration.receiveLoss) {
                    socket.send(ackPocket);
                }else {
                    System.out.println("没有发送ack：" + sequence);
                }
            }
        }
        return byteArrayOutputStream.toByteArray();
    }

    public static void send(DatagramSocket socket, byte[] sendBytes) throws IOException {
        //client端重载send方法，因客户端IP和端口号在服务器端首先发送给客户端时已获得
        send(socket, sendBytes, fromHost, fromPort);
    }
    public static void receiveAndWriteToFile(FileOutputStream fileOutputStream, DatagramSocket socket) throws IOException {
        byte[] receiveBytes;
        while (true) { //使其一直处于接收状态
            receiveBytes = receive(socket);
            if (receiveBytes.length != 0) {
                System.out.println("----------接收成功----------");
                System.out.println("共收到<"+receiveBytes.length+">字节");
                fileOutputStream.write(receiveBytes);
                break;
            }
        }
    }
}

/**
 * 用于对发送的报文进行计时的类
 */
class SRTimeOut extends Thread {

    private final Map<Integer, Integer> ackMap;
    private final DatagramSocket socket;
    private final DatagramPacket packet;
    private final int sequence;

    public SRTimeOut(Map<Integer, Integer> ackMap, DatagramSocket socket, DatagramPacket packet, int sequence) {
        this.ackMap = ackMap;
        this.socket = socket;
        this.packet = packet;
        this.sequence = sequence;
    }

    @Override
    public void run() {
        long startTime=System.currentTimeMillis();   //获取开始时间
        long endTime;
        do {
            try {
                if (ackMap.get(sequence) == null || ackMap.get(sequence) == 1)
                    return;//成功收到ack，直接返回
            }
            catch (Exception e){
//                e.printStackTrace();
            }
            endTime = System.currentTimeMillis(); //获取结束时间
        }while (endTime-startTime<Configuration.timeOut);
        if (ackMap.get(sequence) == null || ackMap.get(sequence) == 1)
            return;//成功收到ack，直接返回
        if (ackMap.get(sequence) == -1) { //未收到ack
            try {
                System.out.println("重新发送序号：" + sequence);//输出重发信息
                socket.send(packet);
            } catch (IOException e) {
                e.printStackTrace();
            }
            new SRTimeOut(ackMap, socket, packet, sequence).start();//重新计时
        }
    }
}
